package com.wei.weblog.mr;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.net.URI;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Locale;
import java.util.UUID;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.wei.weblog.pre.WebLogParser;
import com.wei.wordcount.WordCountJob;
import com.wei.wordcount.WordCountMap;
import com.wei.wordcount.WordCountReduce;

/**
 * 网站日志访问轨迹
 * 
 * @author Administrator
 */
public class WebLogVisitJob {
	static class WebLogVisitMapper extends
			Mapper<LongWritable, Text, Text, WebLogInfo> {

		Text text = new Text();
		WebLogInfo webLogInfo = new WebLogInfo();

		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String string = value.toString();
			String[] split = string.split("\001");
			webLogInfo
					.set(true, split[1], split[2], split[3], split[5],
							split[4], split[6], split[7], split[8], split[9],
							split[10]);
			text.set(webLogInfo.getRemoteIp());
			context.write(text, webLogInfo);
		}
	}

	static class WebLogVisitReduce extends
			Reducer<Text, WebLogInfo, NullWritable, Text> {
		NullWritable key=NullWritable.get();
		Text text=new Text();
		@Override
		protected void reduce(Text ip, Iterable<WebLogInfo> iterable,
				Context context) throws IOException, InterruptedException {
			List<WebLogInfo> webLogInfoList = new ArrayList<>();
			for (WebLogInfo webLogInfo : iterable) {
				WebLogInfo info = new WebLogInfo();
				try {
					BeanUtils.copyProperties(info, webLogInfo);
					webLogInfoList.add(info);
				} catch (Exception e) {
					continue;
				}
			}
			Collections.sort(webLogInfoList, new Comparator<WebLogInfo>() {
				@Override
				public int compare(WebLogInfo o1, WebLogInfo o2) {
					String requestTime2 = o1.getRequestTime();
					String requestTime3 = o2.getRequestTime();
					SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss", Locale.US);
					try {
						Date parse = df.parse(requestTime2);
						Date parse2 = df.parse(requestTime3);
						return parse.compareTo(parse2);
					} catch (ParseException e) {
						return 0;
					}
				}
			});
			String session=UUID.randomUUID().toString();
			int step=1;
			int viewTime=60;
			int size = webLogInfoList.size();
			if (size==1) {
				WebLogInfo webLogInfo = webLogInfoList.get(0);
				StringBuffer buffer=new StringBuffer();
				buffer.append(step+"\001"+60+"\001"+(false)+"\001"+session+"\001");
				buffer.append(webLogInfo.toString());
				text.set(buffer.toString());
				context.write(key, text);
			}
			for (int i = 0; i < size; i++) {
				if (i == 0) {
					continue;
				}
				WebLogInfo webLogInfo0 = webLogInfoList.get(i-1);
				long timeDiff= WebLogParser.timeDiff(webLogInfoList.get(i).getRequestTime(),webLogInfo0.getRequestTime());
				if (timeDiff < 30*60*1000) {//同一个session	
					StringBuffer buffer=new StringBuffer();
					buffer.append(step+"\001"+timeDiff/1000+"\001"+(timeDiff < 30*60*1000)+"\001"+session+"\001");
					buffer.append(webLogInfo0.toString());
					text.set(buffer.toString());
					context.write(key, text);
					step++;
				}else{
					StringBuffer buffer=new StringBuffer();
					buffer.append(step+"\001"+60+"\001"+(timeDiff < 30*60*1000)+"\001"+session+"\001");
					buffer.append(webLogInfo0.toString());
					text.set(buffer.toString());
					context.write(key, text);
					step=1;
					session=UUID.randomUUID().toString();
				}
				
				if (i==size-1) {
					StringBuffer buffer=new StringBuffer();
					buffer.append(step+"\001"+60+"\001"+(timeDiff < 30*60*1000)+"\001"+session+"\001");
					buffer.append(webLogInfoList.get(i).toString());
					text.set(buffer.toString());
					context.write(key, text);
				}
			}

		}
	}
	
	public static void main(String[] args) throws Exception {
		Configuration configuration = new Configuration();
		Job job = Job.getInstance(configuration);
		job.setJarByClass(WebLogVisitJob.class);

		job.setMapperClass(WebLogVisitMapper.class);
		job.setReducerClass(WebLogVisitReduce.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(WebLogInfo.class);

		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(Text.class);
		Path in = new Path(args[0]);
		Path out = new Path(args[1]);
		FileSystem fileSystem = FileSystem.get(new URI(in.toString()), new Configuration());
		 if (fileSystem.exists(out)) {
		      fileSystem.delete(out, true);
		  }
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		boolean waitForCompletion = job.waitForCompletion(true);
		System.exit(waitForCompletion ? 0 : 1);
	}
}
